package hai.mq.kafka;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

import java.util.Random;

public class HaiPartitioner implements Partitioner {

	private final Random ra = new Random();

	public HaiPartitioner(VerifiableProperties props) {

	}

	@Override
	public int partition(Object key, int partitionCount) {
		if (key == null) {
			return ra.nextInt(partitionCount);
		} else {
			return byteArrayToInt(((String) key).getBytes()) % partitionCount;
		}
	}

	private int byteArrayToInt(byte[] b) {
		return b[3] & 0xFF | (b[2] & 0xFF) << 8 | (b[1] & 0xFF) << 16 | (b[0] & 0xFF) << 24;
	}
}